﻿
#include "stdafx.h"
#include "utlist.h"

thrd_comm_queue_t _in_task; // 任务输入队列
thrd_comm_queue_t _out_task;// 任务输出队列
boost::thread *pthrd_switch;//线程指针

static boost::asio::io_service io_service;
static bool stop_flg = false;//线程退出标志


//线程通讯的信息
typedef struct thrd_comm_info
{
    bool valid;
    char name[5];//线程名
    task_addr_t addr;//线程地址
    thrd_ctl_list_node_t *ctl_list_hdr;//接受的命令队列
    thrd_comm_queue_t *_in; //线程输入队列
    thrd_comm_queue_t *_out; //线程输出队列
    boost::condition_variable *_in_cond; //输入条件变量

    unsigned int num_in;//输入数据包数量
    unsigned int num_out;//输出数据包数量
} thrd_comm_info_t;

//用于记录线程的信息
static thrd_comm_info_t thrd_info[10];
static int thrd_num = 0;
static boost::mutex _thrd_info_mutex;//线程互斥量

/*
 *	@brief tcp管理服务器通讯线程——处理任务队列
 */
static void task_handler(thrd_comm_entry_t &task)
{
    switch(task.type)
    {
        /// 停止线程
    case TYP_THRD_STOP:
        switch(task.dir)
        {
        case DIR_REQUEST:
            break;
        case DIR_RESPONSE:
            break;
        case DIR_NO_RESPONSE:
            stop_flg = true;
            break;
        }
        break;
    case TYP_PROG_QUIT://程序退出
        switch(task.dir)
        {
        case DIR_REQUEST:
            break;
        case DIR_RESPONSE:
            break;
        case DIR_NO_RESPONSE:
            task.dst = ADDR_ALL;
            task.type = TYP_THRD_STOP;
            send_task_entry(&task);
            break;
        }
        break;
    default:
        break;
    }
}

/*
 *@brief 线程内部通讯路由函数
 */
static int task_route(thrd_comm_entry_t &task)
{
    int i = 0;
    thrd_comm_queue_t *ptask;
    thrd_ctl_list_node_t * ptr, *tmp;
    ctl_cmd_t cmd;
    bool flg = false;

    boost::mutex::scoped_lock lock(_thrd_info_mutex);
    if(task.ttl == 0)
    {
        //TODO：返回错误-丢弃太长时间没有发出去的数据包。
        return -1;
    }
    for(i = 0;  (thrd_info[i].valid == true) && (i < thrd_num); i++)
    {
        // 查找线程
        if(thrd_info[i].addr == task.src)
        {
            thrd_info[i].num_out ++;//输出数据包计数+1
            break;
        }
    }
    if( i == thrd_num)
    {
        //TODO：返回错误-无效的源地址。
        return -1;
    }
    task.ttl --;
    //cmd = ctl_cmd_t(task.data_buf[4] << 8 + task.data_buf[5]);
    for(i = 0; (thrd_info[i].valid == true) && (i < thrd_num); i++)
    {
        // 查找线程或者是广播通讯
        if((thrd_info[i].addr == task.dst) || (ADDR_ALL == task.dst))
        {
            thrd_info[i].num_in ++;//输入数据包计数+1
            if(thrd_info[i]._in != NULL)//线程有输入管道
            {
                //将通讯写入管道
                ptask = thrd_info[i]._in;
                {
                    boost::mutex::scoped_lock lock(ptask->_mutex);
                    ptask->_queue.push(task);
                    ptask->_cond.notify_one();
                    // 如果注册了条件变量
                    if(thrd_info[i]._in_cond != NULL)
                    {
                        thrd_info[i]._in_cond->notify_one();
                    }
                    flg = true;
                }
                
            }
            else
            {
                //TODO：返回错误-线程没有输入管道。
                return -2;
            }
        }
        else if(ADDR_IN == task.dst)// 输入通讯
        {
            
            LL_FOREACH_SAFE(thrd_info[i].ctl_list_hdr, ptr, tmp)
            {
                if(ptr->cmd == task.type)
                {
                    thrd_info[i].num_in ++;//输入数据包计数+1
                    if(thrd_info[i]._in != NULL)//线程有输入管道
                    {
                        //将通讯写入管道
                        task.dst = (task_addr_t)thrd_info[i].addr;
                        ptask = thrd_info[i]._in;
                        {
                            boost::mutex::scoped_lock lock(ptask->_mutex);
                            ptask->_queue.push(task);
                            ptask->_cond.notify_one();
                            // 如果注册了条件变量
                            if(thrd_info[i]._in_cond != NULL)
                            {
                                thrd_info[i]._in_cond->notify_one();
                            }
                            flg = true;
                        }
                        
                    }
                    else
                    {
                        //TODO：返回错误-线程没有输入管道。
                        return -2;
                    }
                }
            }
        }
        else if((ADDR_OUT == task.dst) && (thrd_info[i].addr == ADDR_TCP))// 输出通讯
        {
            thrd_info[i].num_in ++;//输入数据包计数+1
            if(thrd_info[i]._in != NULL)//线程有输入管道
            {
                //将通讯写入管道
                ptask = thrd_info[i]._in;
                {
                    boost::mutex::scoped_lock lock(ptask->_mutex);
                    ptask->_queue.push(task);
                    ptask->_cond.notify_one();
                    // 如果注册了条件变量
                    if(thrd_info[i]._in_cond != NULL)
                    {
                        thrd_info[i]._in_cond->notify_one();
                    }
                    flg =true;
                }
                
            }
            else
            {
                //TODO：返回错误-线程没有输入管道。
                return -2;
            }
        }
    }
    // 没有找到目的线程
    if(flg == false)
    {
        //向交换线程的输入队列写入，重新进行交换
        boost::mutex::scoped_lock lock(_in_task._mutex);
        _in_task._queue.push(task);
        //_in_task._cond.notify_one();
    }
    return 0;
}
//static UINT timer_id = 10;
/*
 *	@brief 定时器回调函数
 */
//VOID CALLBACK on_timer( HWND, UINT, UINT, DWORD )
//{
//    //向交换线程的输入队列写入，重新进行交换
//    boost::mutex::scoped_lock lock(_in_task._mutex);
//    _in_task._cond.notify_one();
//    return ;
//}
/*
 * @brief 交换线程处理函数
 */
static void switch_thrd_routin()
{
    thrd_comm_entry_t task;
  
    //SetTimer(NULL, timer_id, 1000, on_timer);
    /// 循环直到通知退出线程
    while(stop_flg == false)
    {
        {
            // 查询是否有等待处理的任务
            boost::mutex::scoped_lock lock(_in_task._mutex);
            if(_in_task._queue.size() <= 0 )
            {
                _in_task._cond.wait(lock);
                continue;
            }
            task = _in_task._queue.front();
            _in_task._queue.pop();
        }
        // 是不是发给交换线程的
        if(task.dst == ADDR_SWITCH)
        {
            task_handler(task);
        }
        else
        {
            task_route(task);
        }
    }
    //KillTimer(NULL, timer_id);
    logger_log(LOG_GRP_ALL, LOG_LVL_INFO, "远程管理通讯线程退出！！！");

}

/************************************************************************/
/* @brief 从名字得到地址                                                */
/************************************************************************/
task_addr_t addr_from_name(const char * name)
{
    int i;

    for(i=0; i<8; i++)
    {
        if(0 == strncmp(name, TASK_ADDR_TO_STRING(i), 4))
        {
            return (task_addr_t)i;
        }
    }
}
/*
*@brief 向交换线程注册本线程
*/
int register_thrd(const char *name, thrd_comm_queue_t *_in_queue, thrd_comm_queue_t *_out_queue, boost::condition_variable *cond)
{
    int i;
    if(name == NULL)
    {
        //TODO：返回错误-参数为空。
        return -1;
    }
    boost::mutex::scoped_lock lock(_thrd_info_mutex);
    for(i = 0; (thrd_info[i].valid == true) && (i < thrd_num); i++)
    {
        // 轮询是否有重名线程
        if(memcmp(thrd_info[i].name, name, 4) == 0)
        {
            //TODO：返回错误-有重名线程已经注册
            return -1;
        }
    }
    //添加线程信息
    memcpy(thrd_info[i].name, name, sizeof(thrd_info[i].name) - 1);
    thrd_info[i].addr = addr_from_name(name);
    thrd_info[i]._in = _in_queue;
    thrd_info[i]._out = _out_queue;
    thrd_info[i].valid = true;
    thrd_info[i]._in_cond = cond;
    thrd_num ++;
    char log[256];
    sprintf(log, "%s线程注册成功。地址为：%d、输入管道：%s、输出管道：%s",
            thrd_info[i].name, thrd_info[i].addr,
            (thrd_info[i]._in == NULL ? "无\0" : "有\0"),
            (thrd_info[i]._out == NULL ? "无\0" : "有\0"));

    logger_log(LOG_GRP_ALL, LOG_LVL_INFO, log, strlen(log));

    return 0;
}

/************************************************************************/
/* @brief 向交换线程注册本线程接受的命令                                */
/************************************************************************/
int register_thrd_ctl_cmd(const char *name, thrd_ctl_list_node_t * pnode)
{
    int i;
    thrd_ctl_list_node_t * ptr;
    if(name == NULL)
    {
        //TODO：返回错误-参数为空。
        return -1;
    }
    boost::mutex::scoped_lock lock(_thrd_info_mutex);
    for(i = 0; (thrd_info[i].valid == true) && (i < thrd_num); i++)
    {
        // 轮询是否有重名线程
        if(memcmp(thrd_info[i].name, name, 4) == 0)
        {
            ptr = (thrd_ctl_list_node_t *)CALLOC(1, sizeof(thrd_ctl_list_node_t));
            MEMCPY(ptr, pnode, sizeof(thrd_ctl_list_node_t));
            LL_APPEND(thrd_info[i].ctl_list_hdr, ptr);
            char log[256];
            sprintf(log, "%s线程添加命令=%d()",
                thrd_info[i].name, ptr->cmd, TASK_TYPE_TO_STRING(ptr->cmd));

            logger_log(LOG_GRP_ALL, LOG_LVL_INFO, log, strlen(log));
            break;
        }
    }


    return 0;    
}

/*
 *	@brief 准备并发送内部通讯
 */
int send_task(
    unsigned int dst,                       // 目的线程
    unsigned int src,                       // 源线程
    unsigned int session,                   // 会话ID，可以用于表示一个会话
    task_type_t type,                       // 任务类型
    task_dir_t dir,                         // 任务方向
    unsigned int seqno,                     // 序列号，回复的序列号必须和请求的序列号相同
    unsigned int data_size,                 // 携带的数据长度
    unsigned char*  data_buf                 // 数据区
)
{
    thrd_comm_entry_t task;

    task.dst = (task_addr_t)dst;
    task.src = (task_addr_t)src;
    task.session = session;
    task.type = type;
    task.dir = dir;
    task.seqno = seqno;
    task.ttl = TASK_TTL_MAX;
    task.data_size = data_size;
    if(data_size > sizeof(task.data_buf))
    {
        //TODO：返回错误-数据长度超过缓冲区长度
        return -1;
    }
    memcpy((void *)task.data_buf, (void *)data_buf, data_size);
    
    //向交换线程的输入队列写入
    send_task_entry(&task);
    return 0;
}
/*
 *	
 */
/*
 *	@brief 准备并发送内部通讯
 */
int send_task_entry(thrd_comm_entry_t *ptask)
{
    ptask->ttl = TASK_TTL_MAX;
    
    //向交换线程的输入队列写入
    boost::mutex::scoped_lock lock(_in_task._mutex);
    _in_task._queue.push(*ptask);
    _in_task._cond.notify_one();
    return 0;
}
/*
 *	@brief 启动交换线程
 */
int switch_start()
{
    memset((void *) thrd_info, 0x00, sizeof(thrd_info));
    thrd_num = 0;
    pthrd_switch = new boost::thread(switch_thrd_routin);
    return 0;
}

/*
 *	@brief 停止服务线程
 */
void switch_stop()
{
    delete pthrd_switch;
}
